package edu.gmu.middleware;

import edu.gmu.messages.Client;
import edu.gmu.messages.Connect;
import edu.gmu.messages.Disconnect;
import edu.gmu.messages.Topic;
import java.io.IOException;
import java.nio.ByteBuffer;

/**
 * The RemoteStub sends Messages to the Sink and
 * receives Messages from the Sink.
 * <p>
 * The dispatcher receives events from the Sink(s) (note 
 * that there could be more than one) and then executes the
 * callback Client's method (i.e. DispatchEvent).
 * <p>
 * The result is that there is only one callback
 * thread  regardless of how many Sink's there are, Client's
 * of the Dispatcher will only have one thread for the callbacks.
 * The consequence is that should the callbacks take a long time
 * to complete, the dispatchQueue may overflow due to too many
 * messages received, though likely an OutOfMemoryError will occur
 * first.
 * <p>
 * The dispatcher also works in the opposite direction by taking a
 * Message object and writing to the Sink - though calls are executed
 * on the caller's thread to include marshaling and writing.
 * 
 * @author  James H. Pope
 * @version $Revision:$ $Date:$
 */
public class RemoteStub //implements SinkListener
{   
    public static final int DEFAULT_BUFFER_SIZE = 65536;
    
    private Sink sink                = null;
    private volatile boolean stopped = false;
    
    private Marshaler marshaler      = new MarshalerImpl();
    private Unmarshaler unmarshaler  = new UnmarshalerImpl();
    
    private ByteBuffer in            = null;
    private ByteBuffer out           = null;
    
    private MessageRegistry registry = new MessageRegistry();
    
    private Client client            = null;
    
    /**
     * Creates a new RemoteStub specified values.
     * @param sink
     * @param clientId
     * @param clientName
     */
    public RemoteStub(  String remoteAddress,
                        int remotePort,
                        int clientId,
                        String clientName) throws IOException
    {
        this(   SinkFactory.makeTCPSink(remoteAddress, remotePort),
                DEFAULT_BUFFER_SIZE,
                clientId,
                clientName );
    }
    
    /**
     * Creates a new RemoteStub specified values.
     * @param sink
     * @param clientId
     * @param clientName
     */
    public RemoteStub(  Sink sink,
                        int clientId,
                        String clientName) throws IOException
    {
        this(   sink,
                DEFAULT_BUFFER_SIZE,
                clientId,
                clientName );
    }
    
    /**
     * Creates a new RemoteStub specified values.
     * @param remoteAddress
     * @param remotePort
     * @param bufferSize
     * @param clientId
     * @param clientName
     */
    public RemoteStub(  Sink sink,
                        int bufferSize,
                        int clientId,
                        String clientName) throws IOException
    {
        this.sink = sink;
        this.sink.connect();
        
        this.client = new Client();
        this.client.id = clientId;
        this.client.name = clientName;
        
        /*
         * Register RouterMessage bindings
         */
        this.register(Topic.CONNECT,     Connect.class);
        this.register(Topic.DISCONNECT,  Disconnect.class);
        
        /*
         *  NOTE: BIG ALLOCATIONS HERE
         */
        this.in  = ByteBuffer.allocate( bufferSize );
        // Write has own buffer so cannopt fail due to lack of buffer
        this.out = ByteBuffer.allocate( bufferSize );
    }
    
    /**
     * Registers binding between given id and class
     * @param messageId
     * @param class to bind to messageId
     * @returns the newly registered MessageBinding
     * @throws IllegalArgumentException if id is already bound
     */
    public MessageBinding register( int messageId, Class boundClass )
    {
        return this.registry.register(messageId, boundClass);
    }
    
    /**
     * Performs the following actions - all on the caller's thread.
     * 1. Marshals requestMessage to byteBuffer out
     * 2. Writes out to sink
     * 3. Reads in from sink to get response (assumes remots side responds)
     * 4. Unmarshals response into responseMessage
     * 5. return responseMessage
     * @param topic
     * @param message
     * @throws IOException if the unmashaling/write process was not successful
     * @throws IllegalStateException if no MessageBinding for response
     */
    public Message execute( int requestTopic, Message requestMessage ) throws IOException
    {
        this.send(requestTopic, requestMessage);
        /*
         * Now switch over and read from the sink for the response
         * This is what makes this synchronous
         */
        Message responseMessage = receive();
        return responseMessage;
    }
    
    /**
     * Performs the following actions - all on the caller's thread.
     * 1. Marshals requestMessage to byteBuffer out
     * 2. Writes out to sink
     * 3. Reads in from sink to get response (assumes remots side responds)
     * 4. Unmarshals response into responseMessage
     * 5. return responseMessage
     * @param topic
     * @param message
     * @throws IOException if the mashaling/write process was not successful
     */
    public void send( int requestTopic, Message requestMessage ) throws IOException
    {
        Header requestHeader = new Header();
        requestHeader.setTopic(requestTopic);
        this.marshaler.marshal(requestHeader, requestMessage, out);
        //System.out.print("Writing...");
        //print(out,12);
        this.sink.write(out);
    }
    
    /**
     * Performs the following actions - all on the caller's thread.
     * 1. Reads in from sink to get response (assumes remots side responds)
     * 2. Unmarshals response into responseMessage
     * 3. return responseMessage
     * @throws IOException if the read/unmashaling process was not successful
     * @throws IllegalStateException if no MessageBinding for response
     */
    public Message receive() throws IOException
    {
        /*
         * Now switch over and read from the sink for the response
         * This is what makes this synchronous
         */
        Header responseHeader = this.sink.read(in);

        //System.out.println("Client received...");
        //print( inBuffer, 12 );

        int responseTopic = responseHeader.getTopic();
        MessageBinding messageBinding = registry.getMessageBinding(responseTopic);
        if( messageBinding == null )
        {
            String s = "Cannot locate MessageBinding for "+responseTopic+
                    " have you called register(topic, class)?";
            throw new IllegalStateException(s);
        }
        
        // This will takes a while
        Message responseMessage = unmarshaler.unmarshal(responseHeader,
                                                in,
                                                messageBinding);
        return responseMessage;
    }
    
    
    /**
     * Called from source object when the sink is disconnected.
     * @param event
     */
    public void sinkDisconnected( SinkDisconnectedEvent event )
    {
        // Regardless of disconnect type - go ahead and shutdown
        // Another option is if exception, try to reconnect
        //--------------------------------------------------------------------//
        // Called on Sink readerThread - last statement it executes
        //--------------------------------------------------------------------//
        this.stop();
        //--------------------------------------------------------------------//
    }
    
    
    
    /**
     * Stops the RemoteStub from sending/receiving. Also results in
     * the sink being stopped.
     */
    public void stop( )
    {
        this.stopped = true;
        this.sink.disconnect();
    }
    
    public String toString()
    {
        return "Sink: "+this.sink + "  Client: " + this.client.name;
    }
    
    
    /**
     * Solely to be used for debugging.
     * @param inBuffer
     * @param length
     */
    private static void print( ByteBuffer inBuffer, int length )
    {
        byte[] in = inBuffer.array();
        System.out.print( "position="+inBuffer.position() + " limit="+inBuffer.limit() );
        for (int j = 0; j < length; j++)
        {
            String s = Integer.toHexString( in[j]&0xff );
            if( s.length() == 1 ) s = "0"+s;
            System.out.print( " " + s );
        }
        System.out.println( "" );
    }
    
    


}
